package com.kitten.count.consumer;


import cn.hutool.core.collection.CollectionUtil;
import com.google.common.util.concurrent.RateLimiter;
import com.kitten.count.constant.MQConstants;
import com.kitten.count.domain.mapper.UserCountDOMapper;
import com.kitten.framework.common.utils.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @author kitten
 */
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "kitten_group_" + MQConstants.TOPIC_COUNT_FANS_2_DB,
        topic = MQConstants.TOPIC_COUNT_FANS_2_DB)
public class CountFans2DBConsumer implements RocketMQListener<String> {
    private final UserCountDOMapper userCountDOMapper;
    // 令牌桶削峰
    private RateLimiter rateLimiter = RateLimiter.create(100);  // 每秒100个令牌

    public CountFans2DBConsumer(UserCountDOMapper userCountDOMapper) {
        this.userCountDOMapper = userCountDOMapper;
    }

    @Override
    public void onMessage(String body) {
        rateLimiter.acquire();  // 消费消息时，消耗一个令牌, 若令牌不足则阻塞
        log.info("粉丝计数入库消费 Received message: " + body);
        Map<Long, Integer> countMap = null;
        try {
            countMap = JsonUtils.parseMap(body, Long.class, Integer.class);
        } catch (Exception e) {
            log.error("解析粉丝计数数据失败", e);
        }
        if (CollectionUtil.isNotEmpty(countMap)) {
            countMap.forEach((userId, count) -> {
                userCountDOMapper.insertOrUpdateFansTotalByUserId(userId, count);
            });
        }
    }
}
